package com.wt.springsamples.kafka.listener;


import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wt.springsamples.kafka.Constants;
import com.wt.springsamples.kafka.dto.UserDto;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;


@Component
public class KafkaMessageListener {

    @Autowired
    private  KafkaProperties kafkaProperties;

    @Autowired
    private ObjectMapper objectMapper;

    @KafkaListener(topics = Constants.TOPIC_NAME,groupId = Constants.CONSUMER_GROUP_ID)
    public void listen1(ConsumerRecord<String,String> record, Acknowledgment acknowledgment){
        System.out.println("listen1......"+record.value()+",enableAutoCommit=>"+kafkaProperties.getConsumer().getEnableAutoCommit());
        try {
            objectMapper.readValue("{\"id\":\"1111\",\"name\":\"test22\"}", UserDto.class);
            if(record.value().equals("value10")){
                return;
            }
            TimeUnit.SECONDS.sleep(1);
        } catch (JsonProcessingException | InterruptedException e) {
            throw new RuntimeException(e);
        }
        acknowledgment.acknowledge();
    }

    @KafkaListener(topics = Constants.TOPIC_NAME,groupId = Constants.CONSUMER_GROUP_ID)
    public void listen2(ConsumerRecord<String,String> record,Acknowledgment acknowledgment){
        System.out.println("listen2......"+record.value());
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        /**
         *配置max.poll.interval.ms属性值，表示消费者线程在3秒后属于离线状态会被broker踢出消费者群组，发生再均衡，
         * 后续消费者内调用acknowledgment.acknowledge()会发生CommitFailedException: Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment;
         * spring.kafka.consumer.properties.max.poll.interval.ms=3000
         */
        acknowledgment.acknowledge();
    }

    @KafkaListener(topics = Constants.TOPIC_NAME,groupId = Constants.CONSUMER_GROUP_ID)
    public void listen3(ConsumerRecord<String,String> record,Acknowledgment acknowledgment){
        System.out.println("listen3......"+record.value());
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        acknowledgment.acknowledge();
    }
}
